#!/usr/bin/env python
# -*- coding: utf-8 -*-

import glob
# 导入 glob 模块，用于查找符合特定模式的文件路径名
import json
# 导入 json 模块，用于处理 JSON 数据
import logging
# 导入 logging 模块，用于记录日志
import os
# 导入 os 模块，用于与操作系统进行交互，例如文件路径操作
import shutil
# 导入 shutil 模块，用于文件和文件夹的高级操作，例如移动文件
import time
# 导入 time 模块，用于时间相关的操作，例如暂停执行

from myapp.biz.dataset.api import get_version_info
# 从 myapp.biz.dataset.api 模块导入 get_version_info 函数，用于获取数据集版本信息
from myapp.configs.auto_learning import AUTO_LEARNING_CFGS
# 从 myapp.configs.auto_learning 模块导入 AUTO_LEARNING_CFGS，用于自动学习配置
from myapp.const.dataset import EnumDatasetLabelType, EnumDatasetPublishStatus, EnumDatasetStatus
# 从 myapp.const.dataset 模块导入数据集相关的枚举类型，包括标签类型、发布状态和数据集状态
from myapp.const.user import EnumUserRole
# 从 myapp.const.user 模块导入用户角色枚举类型
from myapp.models.model_dataset import Dataset, DatasetV2
# 从 myapp.models.model_dataset 模块导入 Dataset 和 DatasetV2 模型，用于操作数据集表
from myapp.models.model_eval_dataset import EvalDataset
# 从 myapp.models.model_eval_dataset 模块导入 EvalDataset 模型，用于操作评估数据集表
from myapp.models.model_job import Workflow
# 从 myapp.models.model_job 模块导入 Workflow 模型，用于操作工作流表
from myapp.models.model_model_version import Model_Version
# 从 myapp.models.model_model_version 模块导入 Model_Version 模型，用于操作模型版本表
from myapp.models.model_user import MyUser
# 从 myapp.models.model_user 模块导入 MyUser 模型，用于操作用户表
from myapp.tasks.celery_app import celery_app
# 从 myapp.tasks.celery_app 模块导入 celery_app 实例，用于注册 Celery 任务
from myapp.third.auth.permission import set_roles_for_user_by_id
# 从 myapp.third.auth.permission 模块导入 set_roles_for_user_by_id 函数，用于根据用户 ID 设置角色
from myapp.utils.exception import ignore_exception, log_exception
# 从 myapp.utils.exception 模块导入 ignore_exception 和 log_exception 上下文管理器，用于异常处理
from myapp.utils.sess import scan_table, session_scope
# 从 myapp.utils.sess 模块导入 scan_table 和 session_scope，用于数据库会话管理和表扫描


logger = logging.getLogger(__name__)
# 获取当前模块的日志记录器


@celery_app.task
# 使用 celery_app.task 装饰器将函数注册为 Celery 任务
def conv_dataset():
    """
    转换数据集，将老版微调格式转成新版
    Convert dataset from old fine-tuning format to new format.
    """
    page_size = 1
    # 设置每次查询的数据集数量为 1
    last_id = 0
    # 初始化最后一个处理的数据集 ID 为 0
    while True:
        # 循环处理数据集，直到没有更多符合条件的数据集
        with session_scope() as session:
            # 在数据库会话中执行操作，确保会话的正确管理
            q = (
                session.query(Dataset)
                .filter(
                    Dataset.status == EnumDatasetStatus.succeed.value,
                    # 过滤状态为成功的任务
                    Dataset.labeled == 1,
                    # 过滤已标注的数据集
                    Dataset.label_type == EnumDatasetLabelType.text_generation_sft.value,
                    # 过滤标签类型为文本生成 SFT 的数据集
                    Dataset.data_type == 'txt',
                    # 过滤数据类型为文本的数据集
                    Dataset.id > last_id,
                    # 过滤 ID 大于 last_id 的数据集，用于分页
                )
                .order_by(Dataset.id)
                # 按照 ID 升序排序
            )
            datasets = q.limit(page_size).all()
            # 查询指定数量的数据集
            for dataset in datasets:
                # 遍历查询到的每个数据集
                with log_exception:
                    # 捕获并记录处理单个数据集时可能发生的异常
                    dataset_path = dataset.dataset_path
                    # 获取数据集路径
                    if dataset_path is None:
                        # 如果数据集路径为空，则跳过当前数据集
                        continue
                    if len(dataset_path) == 0:
                        # 如果数据集路径为空字符串，则跳过当前数据集
                        continue
                    if dataset_path == '/':
                        # 如果数据集路径是根目录，则跳过当前数据集
                        continue

                    json_files = glob.glob(
                        os.path.join('/mnt/publish-data', dataset_path, '*.json')
                        # 在指定路径下查找所有 .json 文件
                    )
                    if len(json_files) == 0:
                        # 如果没有找到 .json 文件
                        json_files = glob.glob(
                            os.path.join('/mnt/publish-data', dataset_path, '*.json.new')
                            # 尝试查找 .json.new 文件
                        )
                        for json_file in json_files:
                            # 遍历 .json.new 文件并将其重命名为 .json
                            os.rename(json_file, json_file.replace('.json.new', '.json'))
                        continue
                        # 跳过当前数据集，继续处理下一个

                    for json_file in json_files:
                        # 遍历所有找到的 JSON 文件
                        logger.info(f'converting {json_file}')
                        # 记录日志，表示正在转换文件
                        is_fist_line = True
                        # 标记是否是文件的第一行
                        can_replace = True
                        # 标记是否可以替换原始文件
                        new_json_file = json_file.replace('.json', '.json.new')
                        # 创建一个新的文件名，以 .json.new 结尾
                        with open(json_file) as f:
                            # 打开原始 JSON 文件进行读取
                            with open(new_json_file, 'w') as f2:
                                # 打开新的 JSON 文件进行写入
                                for line in f:
                                    # 逐行读取原始文件内容
                                    line = line.strip()
                                    # 移除行首尾空白字符
                                    if line is None:
                                        # 如果行内容为空，则跳过
                                        continue
                                    if len(line) == 0:
                                        # 如果行内容为空字符串，则跳过
                                        continue

                                    new_conversations = [
                                        {
                                            'role': 'system',
                                            'content': '你是紫东太初，也叫小初，'
                                            '你是由武汉人工智能研究院和中国科学院自动化研究所联合研发的千亿参数多模态大模型.\n',
                                        }
                                    ]
                                    # 初始化新的对话列表，包含一个系统角色消息
                                    with ignore_exception:
                                        # 忽略处理 JSON 数据时可能发生的异常
                                        data = json.loads(line)
                                        # 解析 JSON 格式的行数据
                                        if data is None:
                                            # 如果解析数据为空，则跳过
                                            continue

                                        conversations = data.get('conversations', [])
                                        # 获取对话列表，如果不存在则为空列表
                                        if len(conversations) == 0:
                                            # 如果对话列表为空，则记录日志并跳过当前文件
                                            logger.info('conversations is empty, skip')
                                            break

                                        if is_fist_line:
                                            # 如果是文件的第一行
                                            is_fist_line = False
                                            # 将第一行标记设置为 False
                                            conversation = conversations[0]
                                            # 获取第一个对话
                                            from_ = conversation.get('from')
                                            # 获取对话的来源
                                            if from_ is None:
                                                # 如果来源为空，则标记为不可替换并跳过
                                                can_replace = False
                                                break

                                            if from_ not in ['question', 'answer']:
                                                # 如果来源不是 'question' 或 'answer'，则标记为不可替换并跳过
                                                can_replace = False
                                                break

                                        for conversation in conversations:
                                            # 遍历对话列表中的每个对话
                                            from_ = conversation.get('from', '')
                                            # 获取对话的来源
                                            if from_ == 'question':
                                                # 如果来源是 'question'，则添加用户角色消息
                                                new_conversations.append(
                                                    {
                                                        'role': 'user',
                                                        'content': conversation.get('value', ''),
                                                    }
                                                )
                                            if from_ == 'answer':
                                                # 如果来源是 'answer'，则添加助手角色消息
                                                new_conversations.append(
                                                    {
                                                        'role': 'assistant',
                                                        'content': conversation.get('value', ''),
                                                    }
                                                    # Append an assistant role message if the source is 'answer'
                                                )

                                    if len(new_conversations) > 1:
                                        # 如果新的对话列表包含多条消息
                                        f2.write(
                                            json.dumps(
                                                {'conversations': new_conversations},
                                                ensure_ascii=False,
                                            )
                                            # 将新的对话列表转换为 JSON 字符串，并确保非 ASCII 字符正确编码
                                            + '\n'
                                            # 添加换行符
                                        )
                        if can_replace:
                            # 如果可以替换原始文件
                            os.remove(json_file)
                            # 删除原始 JSON 文件
                            time.sleep(1)
                            # 暂停 1 秒
                            os.rename(new_json_file, json_file)
                            # 将新文件重命名为原始文件名
                        else:
                            # 如果不能替换原始文件
                            os.remove(new_json_file)
                            # 删除新创建的 .json.new 文件

            if len(datasets) < page_size:
                # 如果查询到的数据集数量小于每页大小，说明已经处理完所有数据，跳出循环
                break
            last_id = datasets[-1].id
            # 更新 last_id 为最后一个处理的数据集的 ID，用于下一页查询


@celery_app.task
# 使用 celery_app.task 装饰器将函数注册为 Celery 任务
def migrate_dataset():
    """
    迁移数据集
    Migrate datasets.
    """
    page_size = 100
    # 设置每次查询的数据集数量为 100
    with session_scope() as session:
        # 在数据库会话中执行操作
        q = session.query(Dataset).filter_by(dataset_id=None)
        # 查询 dataset_id 为 None 的数据集
        while True:
            # 循环处理数据集
            datasets = q.limit(page_size).all()
            # 查询指定数量的数据集
            for dataset in datasets:
                # 遍历查询到的每个数据集
                with log_exception:
                    # 捕获并记录处理单个数据集时可能发生的异常
                    if dataset.status != 'succeed':
                        # 如果数据集状态不是 'succeed'，则记录日志并跳过
                        logger.info(f'数据集 {dataset.name} 状态为 {dataset.status}，跳过')
                        continue
                    if dataset.labeled != 1:
                        # 如果数据集未标注，则记录日志并跳过
                        logger.info(f'数据集 {dataset.name} 未标注，跳过')
                        continue

                    dataset_v2 = DatasetV2()
                    # 创建一个新的 DatasetV2 实例
                    dataset_v2.name = dataset.name
                    # 复制数据集名称
                    dataset_v2.data_type = dataset.data_type
                    # 复制数据类型
                    dataset_v2.label_type = (
                        '' if dataset.label_type is None else dataset.label_type
                        # 复制标签类型，如果为空则设置为 ''
                    )
                    dataset_v2.source = dataset.source
                    # 复制数据来源
                    dataset_v2.describe = '' if dataset.describe is None else dataset.describe
                    # 复制描述，如果为空则设置为 ''
                    dataset_v2.created_by_fk = dataset.created_by_fk
                    # 复制创建者外键
                    dataset_v2.changed_by_fk = dataset.changed_by_fk
                    # 复制修改者外键
                    dataset_v2.created_on = dataset.created_on
                    # 复制创建时间
                    dataset_v2.changed_on = dataset.changed_on
                    # 复制修改时间
                    if dataset.source == 'internal':
                        # 如果数据来源是 'internal'
                        dataset_v2.created_by_fk = 0
                        # 将创建者外键设置为 0
                        dataset_v2.changed_by_fk = 0
                        # 将修改者外键设置为 0
                        dataset.created_by_fk = 0
                        # 将原始数据集的创建者外键设置为 0
                        dataset.changed_by_fk = 0
                        # 将原始数据集的修改者外键设置为 0

                    session.add(dataset_v2)
                    # 将新的 DatasetV2 实例添加到会话
                    session.flush()
                    # 刷新会话，使 dataset_v2.id 可用
                    dataset.dataset_id = dataset_v2.id
                    # 将原始数据集的 dataset_id 设置为新的 DatasetV2 的 ID
                    dataset.pub_status = EnumDatasetPublishStatus.published.value
                    # 将原始数据集的发布状态设置为 'published'

                    logger.info(f'迁移数据集 {dataset.name} 到 {dataset_v2.id}')
                    # 记录日志，表示数据集迁移成功
                    session.commit()
                    # 提交数据库事务

            if len(datasets) < page_size:
                # 如果查询到的数据集数量小于每页大小，说明已经处理完所有数据，跳出循环
                break


@celery_app.task
# 使用 celery_app.task 装饰器将函数注册为 Celery 任务
def fix_internal_dataset():
    """
    修复内置数据集
    Fix internal datasets.
    """
    page_size = 1
    # 设置每次查询的数据集数量为 1
    last_id = 0
    # 初始化最后一个处理的数据集 ID 为 0
    while True:
        # 循环处理数据集
        with session_scope() as session:
            # 在数据库会话中执行操作
            q = (
                session.query(Dataset)
                .filter(
                    Dataset.status == EnumDatasetStatus.succeed.value,
                    # 过滤状态为成功的任务
                    Dataset.labeled == 1,
                    # 过滤已标注的数据集
                    Dataset.label_type.in_(
                        [
                            EnumDatasetLabelType.text_generation_sft.value,
                            EnumDatasetLabelType.text_generation_pretrain.value,
                        ]
                    ),
                    # 过滤标签类型为文本生成 SFT 或文本生成预训练的数据集
                    Dataset.data_type == 'txt',
                    # 过滤数据类型为文本的数据集
                    Dataset.source == 'internal',
                    # 过滤数据来源为 'internal' 的数据集
                    Dataset.id > last_id,
                    # 过滤 ID 大于 last_id 的数据集，用于分页
                )
                .order_by(Dataset.id)
                # 按照 ID 升序排序
            )
            datasets = q.limit(page_size).all()
            # 查询指定数量的数据集
            for dataset in datasets:
                # 遍历查询到的每个数据集
                with log_exception:
                    # 捕获并记录处理单个数据集时可能发生的异常
                    dataset_path = dataset.dataset_path
                    # 获取数据集路径
                    if dataset_path is None:
                        # 如果数据集路径为空，则跳过当前数据集
                        continue
                    if len(dataset_path) == 0:
                        # 如果数据集路径为空字符串，则跳过当前数据集
                        continue
                    if dataset_path == '/':
                        # 如果数据集路径是根目录，则跳过当前数据集
                        continue

                    # if dataset.entries_num is not None and len(dataset.entries_num) > 0:
                    # # 如果数据集的条目数不为空且大于 0，则跳过
                    #     continue

                    json_files = glob.glob(
                        os.path.join('/mnt/publish-data', dataset_path, '*.json')
                        # 在指定路径下查找所有 .json 文件
                    )
                    if len(json_files) == 0:
                        # 如果没有找到 .json 文件，则跳过
                        continue

                    num = 0
                    # 初始化条目计数器
                    json_file = json_files[0]
                    # 获取第一个 JSON 文件
                    with open(json_file) as f:
                        # 打开 JSON 文件进行读取
                        for line in f:
                            # 逐行读取文件内容
                            line = line.strip()
                            # 移除行首尾空白字符
                            if line is None or len(line) == 0:
                                # 如果行内容为空或空字符串，则跳过
                                continue

                            data = None
                            # 初始化数据为 None
                            with ignore_exception:
                                # 忽略处理 JSON 数据时可能发生的异常
                                data = json.loads(line)
                                # 解析 JSON 格式的行数据
                            if data is None:
                                # 如果解析数据为空，则跳过
                                continue

                            num += 1
                            # 增加条目计数

                    dataset.entries_num = str(num)
                    # 更新数据集的条目数
                    session.commit()
                    # 提交数据库事务

            if len(datasets) < page_size:
                # 如果查询到的数据集数量小于每页大小，说明已经处理完所有数据，跳出循环
                break

            last_id = datasets[-1].id
            # 更新 last_id 为最后一个处理的数据集的 ID，用于下一页查询


@celery_app.task
# 使用 celery_app.task 装饰器将函数注册为 Celery 任务
def add_deploy_cfgs_in_model_version():
    """
    刷模型版本的deploy_cfgs字段
    Populate the deploy_cfgs field for model versions.
    """
    page_size = 1000
    # 设置每次查询的模型版本数量为 1000
    last_id = 0
    # 初始化最后一个处理的模型版本 ID 为 0
    while True:
        # 循环处理模型版本
        with session_scope() as session:
            # 在数据库会话中执行操作
            q = (
                session.query(Model_Version)
                .filter(Model_Version.id > last_id)
                # 过滤 ID 大于 last_id 的模型版本，用于分页
                .order_by(Model_Version.id)
                # 按照 ID 升序排序
            )
            mvs = q.limit(page_size).all()
            # 查询指定数量的模型版本
            for mv in mvs:
                # 遍历查询到的每个模型版本
                mv.deploy_cfgs = '{}'
                # 初始化 deploy_cfgs 字段为空 JSON 字符串
                with log_exception:
                    # 捕获并记录处理单个模型版本时可能发生的异常
                    source_id = mv.source_id
                    # 获取模型版本的源 ID
                    if not source_id:
                        # 如果源 ID 不存在，则记录日志并跳过
                        logger.info(f'模型版本 {mv.id} 没有source_id')
                        continue

                    workflow = session.query(Workflow).filter_by(id=source_id).first()
                    # 根据源 ID 查询对应的工作流
                    if not workflow:
                        # 如果工作流不存在，则记录日志并跳过
                        logger.info(f'模型版本 {mv.id} 对应的workflow {source_id} 不存在')
                        continue

                    algorithm = json.loads(workflow.algorithm)
                    # 解析工作流的算法配置
                    task_type = json.loads(workflow.labels).get('task_type')
                    # 从工作流标签中获取任务类型
                    if task_type is None:
                        # 如果任务类型为空，则记录日志并跳过
                        logger.info(f'模型版本 {mv.id} 对应的workflow {source_id} 没有task_type')
                        continue
                    resource = algorithm.get('resource', {})
                    # 获取算法配置中的资源信息
                    resource_type = algorithm.get('resource', {}).get('resource_type', '')
                    # 获取资源类型
                    pool_type = resource.get('pool_type')
                    # 获取池类型
                    lora = algorithm.get('resource', {}).get('lora')
                    # 获取 LoRA 配置
                    deploy_cfgs = {'pool_type': pool_type, 'task_type': task_type}
                    # 初始化部署配置
                    all_cfgs = AUTO_LEARNING_CFGS.get(task_type)
                    # 获取所有自动学习配置中对应任务类型的配置
                    if all_cfgs and resource_type:
                        # 如果存在自动学习配置和资源类型
                        envs = all_cfgs.get(resource_type, {}).get('envs')
                        # 获取环境变量配置
                        specifications = all_cfgs.get(resource_type, {}).get('specifications')
                        # 获取规格配置
                        if lora:
                            # 如果存在 LoRA 配置
                            envs = all_cfgs.get(resource_type, {}).get('lora', {}).get('envs')
                            # 获取 LoRA 相关的环境变量配置
                            specifications = (
                                all_cfgs.get(resource_type, {})
                                .get('lora', {})
                                .get('specifications')
                                # 获取 LoRA 相关的规格配置
                            )
                        deploy_cfgs['envs'] = envs
                        # 将环境变量添加到部署配置
                        deploy_cfgs['specifications'] = specifications
                        # 将规格添加到部署配置
                    mv.deploy_cfgs = json.dumps(deploy_cfgs, ensure_ascii=False)
                    # 将部署配置转换为 JSON 字符串并存储
                    mv.device_type = resource_type
                    # 更新模型版本的设备类型

            session.commit()
            # 提交数据库事务
            if len(mvs) < page_size:
                # 如果查询到的模型版本数量小于每页大小，说明已经处理完所有数据，跳出循环
                break

            last_id = mvs[-1].id
            # 更新 last_id 为最后一个处理的模型版本的 ID，用于下一页查询


@celery_app.task
# 使用 celery_app.task 装饰器将函数注册为 Celery 任务
def fix_sft_file_ext():
    """
    修复内置数据集
    Fix internal datasets.
    """
    page_size = 1
    # 设置每次查询的数据集数量为 1
    last_id = 0
    # 初始化最后一个处理的数据集 ID 为 0
    while True:
        # 循环处理数据集
        with session_scope() as session:
            # 在数据库会话中执行操作
            q = (
                session.query(Dataset)
                .filter(
                    Dataset.status == EnumDatasetStatus.succeed.value,
                    # 过滤状态为成功的任务
                    Dataset.label_type.in_([EnumDatasetLabelType.text_generation_sft.value]),
                    # 过滤标签类型为文本生成 SFT 的数据集
                    Dataset.data_type == 'txt',
                    # 过滤数据类型为文本的数据集
                    Dataset.id > last_id,
                    # 过滤 ID 大于 last_id 的数据集，用于分页
                )
                .order_by(Dataset.id)
                # 按照 ID 升序排序
            )
            datasets = q.limit(page_size).all()
            # 查询指定数量的数据集
            for dataset in datasets:
                # 遍历查询到的每个数据集
                with log_exception:
                    # 捕获并记录处理单个数据集时可能发生的异常
                    dataset_path = dataset.dataset_path
                    # 获取数据集路径
                    if dataset_path is None:
                        # 如果数据集路径为空，则跳过当前数据集
                        continue
                    if len(dataset_path) == 0:
                        # 如果数据集路径为空字符串，则跳过当前数据集
                        continue
                    if dataset_path == '/':
                        # 如果数据集路径是根目录，则跳过当前数据集
                        continue

                    json_files = glob.glob(
                        os.path.join('/mnt/publish-data', dataset_path, '*.json')
                        # 在指定路径下查找所有 .json 文件
                    )
                    if len(json_files) == 0:
                        # 如果没有找到 .json 文件，则跳过
                        continue

                    json_file = json_files[0]
                    # 获取第一个 JSON 文件
                    shutil.move(json_file, ''.join(json_file.split('.')[:-1]) + '.jsonl')
                    # 将 .json 文件重命名为 .jsonl 文件

            if len(datasets) < page_size:
                # 如果查询到的数据集数量小于每页大小，说明已经处理完所有数据，跳出循环
                break

            last_id = datasets[-1].id
            # 更新 last_id 为最后一个处理的数据集的 ID，用于下一页查询


@celery_app.task
# 使用 celery_app.task 装饰器将函数注册为 Celery 任务
def fix_fm_val_dataset_cfg():
    # 修复 FM 验证数据集配置
    # Fix FM validation dataset configuration
    for workflows, sess in scan_table(Workflow):
        # 遍历所有工作流及其对应的数据库会话
        for workflow in workflows:
            # 遍历每个工作流
            with log_exception:
                # 捕获并记录处理单个工作流时可能发生的异常
                algorithm = json.loads(workflow.algorithm)
                # 解析工作流的算法配置
                val_dataset_cfg = algorithm.get('val_dataset_cfg')
                # 获取验证数据集配置
                if val_dataset_cfg is None:
                    # 如果验证数据集配置为空，则跳过
                    continue
                source = val_dataset_cfg.get('source')
                # 获取验证数据集的来源
                if source is None:
                    # 如果来源为空，则跳过
                    continue
                if source == 'dataset':
                    # 如果来源是 'dataset'
                    dataset_version_id = val_dataset_cfg.get('dataset_version_id')
                    # 获取数据集版本 ID
                    if dataset_version_id is None:
                        # 如果数据集版本 ID 为空，则跳过
                        continue
                    dataset_version = get_version_info(dataset_version_id, sess, 1, is_admin=True)
                    # 获取数据集版本信息
                    algorithm['val_dataset_cfg'] = {
                        'source': 'dataset',
                        # 设置来源为 'dataset'
                        'datasets': [
                            {
                                'id': dataset_version.get('dataset_id'),
                                # 设置数据集 ID
                                'name': dataset_version.get('dataset_name'),
                                # 设置数据集名称
                                'version_id': dataset_version_id,
                                # 设置版本 ID
                                'version_name': dataset_version.get('name'),
                                # 设置版本名称
                            }
                        ],
                    }
                    workflow.algorithm = json.dumps(algorithm, ensure_ascii=False)
                    # 将更新后的算法配置转换为 JSON 字符串并存储

        sess.commit()
        # 提交数据库事务


@celery_app.task
# 使用 celery_app.task 装饰器将函数注册为 Celery 任务
def eval_dataset_json_to_jsonl():
    """
    修复内置数据集
    Fix internal datasets.
    """
    for datasets, _ in scan_table(EvalDataset):
        # 遍历所有评估数据集及其对应的数据库会话
        for dataset in datasets:
            # 遍历每个评估数据集
            with log_exception:
                # 捕获并记录处理单个数据集时可能发生的异常
                dataset_path = dataset.dataset_path
                # 获取数据集路径
                if dataset_path is None:
                    # 如果数据集路径为空，则跳过当前数据集
                    continue
                if len(dataset_path) == 0:
                    # 如果数据集路径为空字符串，则跳过当前数据集
                    continue
                if dataset_path == '/':
                    # 如果数据集路径是根目录，则跳过当前数据集
                    continue

                json_files = glob.glob(os.path.join('/mnt/publish-data', dataset_path, '*.json'))
                # 在指定路径下查找所有 .json 文件
                if len(json_files) == 0:
                    # 如果没有找到 .json 文件，则跳过
                    continue

                json_file = json_files[0]
                # 获取第一个 JSON 文件
                shutil.move(json_file, ''.join(json_file.split('.')[:-1]) + '.jsonl')
                # 将 .json 文件重命名为 .jsonl 文件


@celery_app.task
# 使用 celery_app.task 装饰器将函数注册为 Celery 任务
def init_tenant_data():
    """
    多租户刷库
    Multi-tenant database refresh.
    """

    def get_role(arr):
        # 定义一个辅助函数，用于获取用户角色
        # Define a helper function to get user roles
        if arr is None or len(arr) == 0:
            # 如果角色数组为空，则返回 'Common' 角色
            # If the role array is empty, return 'Common' role
            return 'Common'
        all_role = {'Common': 1, 'Senior': 2, 'Vip': 3, 'Admin': 4}
        # 定义所有角色及其对应的优先级
        # Define all roles and their corresponding priorities
        role = max(arr, key=lambda x: all_role.get(x, 0))
        # 获取优先级最高的角色
        # Get the role with the highest priority
        if not all_role.get(role):
            # 如果获取到的角色不在已知角色列表中，则默认为 'Common' 角色
            # If the obtained role is not in the known role list, default to 'Common' role
            role = 'Common'
        return role
        # 返回最终确定的角色

    for users, sess in scan_table(MyUser):
        # 遍历所有用户及其对应的数据库会话
        for u in users:
            # 遍历每个用户
            with log_exception:
                # 捕获并记录处理单个用户时可能发生的异常
                if u.tenant_id is None or len(u.tenant_id) == 0:
                    # 如果用户的 tenant_id 为空，则将其设置为用户的 username
                    # If the user's tenant_id is empty, set it to the user's username
                    u.tenant_id = u.username

                if u.taichu == '':
                    # 如果用户的 taichu 字段为空，则跳过
                    # If the user's taichu field is empty, skip
                    continue

                roles = []
                # 初始化角色列表
                if u.tenant_id == u.username:
                    # 如果 tenant_id 等于 username，则添加 TenantAdmin 角色
                    # If tenant_id equals username, add TenantAdmin role
                    roles.append(EnumUserRole.TenantAdmin.value)
                all_roles = [str(u) for u in u.roles]
                # 获取用户的所有角色并转换为字符串列表
                roles.append(get_role(all_roles))
                # 添加通过 get_role 函数确定的角色

                set_roles_for_user_by_id(u.taichu, list(set(roles)))
                # 根据用户的 taichu ID 设置角色，并去除重复角色

        sess.commit()
        # 提交数据库事务

    # # 以下是被注释掉的代码，用于为其他表初始化 tenant_id
    # # The following commented-out code is used to initialize tenant_id for other tables
    # for table in [
    #     Labels,
    #     Prompt,
    #     Dataset,
    #     DatasetV2,
    #     EvalDataset,
    #     EvaluationTask,
    #     AutoLearning,
    #     Training_Model,
    #     Model_Version,
    #     Notebook,
    #     Codeset,
    #     Service,
    # ]:
    #     for data, sess in scan_table(table):
    #         for item in data:
    #             with log_exception:
    #                 if item.tenant_id is None or len(item.tenant_id) == 0:
    #                     item.tenant_id = item.created_by.tenant_id
    #         sess.commit()